package net.siufung.boot.elasticsearch.service.impl;

import cn.hutool.core.util.ObjectUtil;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import net.siufung.boot.elasticsearch.service.IElasticsearchIndexService;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.admin.indices.alias.Alias;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkItemResponse.Failure;
import org.elasticsearch.action.support.IndicesOptions;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.client.indices.GetIndexResponse;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.ReindexRequest;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.List;
import java.util.Map;

/**
 * @author 陈建峰
 * @since 2022/6/16 2:57 下午
 */
@Component("elasticsearchIndexService")
@AllArgsConstructor
@Slf4j
public class ElasticsearchIndexServiceImpl implements IElasticsearchIndexService {

    private final RestHighLevelClient restHighLevelClient;

    @Override
    public Boolean existsIndex(String indexName) {
        GetIndexRequest request = new GetIndexRequest(indexName);
        try {
            return restHighLevelClient.indices().exists(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            throw new ElasticsearchException("exists Index exception indexName: %s", indexName);
        }
    }

    @Override
    public Boolean deleteIndex(String indexName) {
        DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(indexName);
        deleteIndexRequest.indicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN);
        try {
            AcknowledgedResponse acknowledgedResponse = restHighLevelClient.indices().delete(deleteIndexRequest, RequestOptions.DEFAULT);
            return acknowledgedResponse.isAcknowledged();
        } catch (IOException e) {
            throw new ElasticsearchException("delete Index exception indexName: %s", indexName);
        }
    }

    @Override
    public Boolean createEmptyIndex(String indexName) {
        CreateIndexRequest request = new CreateIndexRequest(indexName);
        CreateIndexResponse createIndexResponse;
        try {
            createIndexResponse = restHighLevelClient.indices().create(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.info("===> distribute lock index has created");
            return Boolean.TRUE;
        }
        return createIndexResponse.isAcknowledged();
    }

    @Override
    public Boolean createIndex(String indexName, Map<String, Object> mapping, Settings settings, String aliasName) {
        CreateIndexRequest createIndexRequest = new CreateIndexRequest(indexName);
        createIndexRequest.settings(settings);
        createIndexRequest.mapping(mapping);
        if(!Strings.isNullOrEmpty(aliasName)){
            createIndexRequest.alias(new Alias(aliasName));
        }
        try {
            CreateIndexResponse createIndexResponse = restHighLevelClient.indices().create(createIndexRequest, RequestOptions.DEFAULT);
            return createIndexResponse.isAcknowledged();
        } catch (IOException e) {
            throw new ElasticsearchException("create index exception createIndexRequest: %s ", e, createIndexRequest.toString());
        }
    }

    @Override
    public GetIndexResponse getIndex(String indexName) {
        GetIndexRequest request = new GetIndexRequest(indexName);
        try {
            return restHighLevelClient.indices().get(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            throw new ElasticsearchException("getIndex exception indexName: %s", e, indexName);
        }
    }

    @Override
    public void addDefaultAlias(String indexName) {
        IndicesAliasesRequest indicesAliasesRequest = new IndicesAliasesRequest();
        IndicesAliasesRequest.AliasActions aliasActions =
                new IndicesAliasesRequest.AliasActions(IndicesAliasesRequest.AliasActions.Type.ADD);
        aliasActions.index(indexName);
        aliasActions.alias("default_alias");
        indicesAliasesRequest.addAliasAction(aliasActions);
        try {
            restHighLevelClient.indices().updateAliases(indicesAliasesRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.warn("add defaultAlias exception:, {}", e.toString());
        }
    }

    @Override
    public Boolean reindex(String oldIndexName, String releaseIndexName) {
        ReindexRequest reindexRequest = new ReindexRequest();
        reindexRequest.setSourceIndices(oldIndexName);
        reindexRequest.setDestIndex(releaseIndexName);
        reindexRequest.setDestOpType("create");
        reindexRequest.setConflicts("proceed");
        reindexRequest.setRefresh(Boolean.TRUE);
        try {
            BulkByScrollResponse response = restHighLevelClient.reindex(reindexRequest, RequestOptions.DEFAULT);
            List<Failure> bulkFailures = response.getBulkFailures();
            if (ObjectUtil.isEmpty(bulkFailures)) {
                return Boolean.TRUE;
            }
            return Boolean.FALSE;
        } catch (IOException e) {
            throw new ElasticsearchException("reindex exception oldIndexName:%s, releaseIndexName: %s", e, oldIndexName, releaseIndexName);
        }
    }

    @Override
    public Boolean changeAliasAtomic(String oldIndexName, String releaseIndexName) {
        IndicesAliasesRequest.AliasActions addIndexAction = new IndicesAliasesRequest.AliasActions(
                IndicesAliasesRequest.AliasActions.Type.ADD).index(releaseIndexName).alias("default_alias");
        IndicesAliasesRequest.AliasActions removeAction = new IndicesAliasesRequest.AliasActions(
                IndicesAliasesRequest.AliasActions.Type.REMOVE).index(oldIndexName).alias("default_alias");

        IndicesAliasesRequest indicesAliasesRequest = new IndicesAliasesRequest();
        indicesAliasesRequest.addAliasAction(addIndexAction);
        indicesAliasesRequest.addAliasAction(removeAction);
        try {
            AcknowledgedResponse acknowledgedResponse = restHighLevelClient.indices().updateAliases(indicesAliasesRequest,
                    RequestOptions.DEFAULT);
            return acknowledgedResponse.isAcknowledged();
        } catch (IOException e) {
            throw new ElasticsearchException("changeAlias exception oldIndexName: %s, releaseIndexName: %s", e, oldIndexName, releaseIndexName);
        }
    }


}
